package com.example.jobback.websocket.service;

import com.alibaba.fastjson2.JSON;
import com.example.jobback.po.Company;
import com.example.jobback.po.Recruiter;
import com.example.jobback.po.User;
import com.example.jobback.service.common.UserCommonService;
import com.example.jobback.service.recruiter.RecruiterCrossService;
import com.example.jobback.util.CACHEUtil;
import com.example.jobback.websocket.common.MessageType;
import com.example.jobback.websocket.common.UserRole;
import com.example.jobback.websocket.dto.MessageSendDto;
import com.example.jobback.websocket.mapper.MessageMapper;
import com.example.jobback.websocket.po.Message;
import com.example.jobback.websocket.resp.ErrorMessageResp;
import com.example.jobback.websocket.resp.LoginMessagesResp;
import com.example.jobback.websocket.resp.MessageSendResp;
import com.example.jobback.websocket.resp.UserRelatedMessageResp;
import jakarta.annotation.PostConstruct;
import jakarta.websocket.CloseReason;
import jakarta.websocket.EndpointConfig;
import jakarta.websocket.Session;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.example.jobback.websocket.util.RedisStreamConstants.UNREAD_MESSAGE_CONSUMER_GROUP;
import static com.example.jobback.websocket.util.RedisStreamConstants.UNREAD_MESSAGE_STREAM;
import static java.time.LocalDateTime.now;
/**
 * @ClassName: MessageService
 * @Description: 消息服务类
 * @Author: lhb
 * @Date: 2025/5/28
 */
@Service
@Slf4j
public class MessageService {

    @Value("${ws.role.seeker}") public  String seekerRole;

    @Value("${ws.role.recruiter}") public  String recruiterRole;

    @Value("${ws.role.manager}") public  String managerRole;

    private final MessageMapper messageMapper;

    private static final Map<Long, Session> onlineUsers = new ConcurrentHashMap<>();

    private static final Map<Long,String> onlineRoles=new ConcurrentHashMap<>();

    private final RecruiterCrossService recruiterCrossService;

    private final UserCommonService userCommonService;

    private final CACHEUtil cacheUtil;


    public MessageService(MessageMapper messageMapper, RecruiterCrossService recruiterCrossService, UserCommonService userCommonService, CACHEUtil cacheUtil) {
        this.messageMapper = messageMapper;
        this.recruiterCrossService = recruiterCrossService;
        this.userCommonService = userCommonService;
        this.cacheUtil = cacheUtil;
    }

    /**
     * @Description: 初始化消费者组 
     * @return void
     * @Author lhb
     * @CreateTime 2025/9/20 22:02
     */
    @PostConstruct
    public void initStreamConsumerGroup() {
        try {
            // 检查并创建消费者组
            cacheUtil.createStreamGroup(UNREAD_MESSAGE_STREAM, UNREAD_MESSAGE_CONSUMER_GROUP);
        } catch (Exception e) {
            log.error("初始化未读消息Stream消费者组失败", e);
        }
    }

    /**
     * @Description: 发送消息成功后，添加未读记录到Redis Stream
     * @param dto
     * @return void
     * @Author lhb
     * @CreateTime 2025/9/20 22:03
     */
    private void addUnreadRecordToStream(MessageSendDto dto) {
        try {
            Map<String, String> record = new HashMap<>();
            record.put("fromUserId", dto.getFromUserId().toString());
            record.put("fromRole", dto.getFromRole());
            record.put("toUserId", dto.getToUserId().toString());
            record.put("toRole", dto.getToRole());
            record.put("createdAt",dto.getCreatedAt().toString());
            // 添加到Redis Stream
            cacheUtil.addStreamRecord(UNREAD_MESSAGE_STREAM, record);
        } catch (Exception e) {
            log.error("添加未读消息记录到Stream失败", e);
        }
    }



    /**
     * @Description: 用户建立websocket连接触发的方法
     * @param session
     * @param config
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:36
     */
    public void onOpen(Session session, EndpointConfig config) throws IOException {
        //发送给客户端error信息
        if (handleHandshakeError(session, config)) return;
        String userIdStr = (String) config.getUserProperties().get("userIdStr");
        String role = (String) config.getUserProperties().get("role");
        log.info("用户开始连接-"+role+"-"+userIdStr);
        long userId = Long.parseLong(userIdStr);
        // 关闭旧的连接，token已失效的连接
        deleteOldSession(userId,session);
        //添加用户到map集合
        registerUserSession(userId, session, role);
        //从mysql获取全量信息
        List<Message> messages = messageMapper.getMessagesByUserId(userId,role);
        if (!messages.isEmpty()) {
            //排序，构造聊天记录
            List<LoginMessagesResp> loginMessagesRespList = processMessages(messages, userId, role);
            //转换成统一的resp格式messageSendResp，发送给客户端
            sendLoginMessages(session, userId, role, loginMessagesRespList);
            //异步操作，根据userId，role，在mapper层把该用户的所有消息标记为在线消息，因为该用户已经登录
            // 异步标记消息为在线，取代
//            markMessagesAsOnline(userId, role);
        }
        //广播在线用户
        broadcastOnlineUsers();
    }

    public void broadcastOnlineUsers(){
        MessageSendResp messageSendResp=new MessageSendResp();
        messageSendResp.setOnline(true);
        messageSendResp.setToUserId(-1L);
        messageSendResp.setFromUserId(-1L);
        messageSendResp.setMessage(onlineRoles);
        messageSendResp.setFromRole("server");
        messageSendResp.setToRole("server");
        messageSendResp.setCreatedAt(now());
        messageSendResp.setMessageType(MessageType.ONLINE_USERS);
        String messageSendRespJson = JSON.toJSONString(messageSendResp);
        for (Session session : onlineUsers.values()) {
            sendTextWithRetry(session, messageSendRespJson);
        }
    }



    /**
     * @Description: 加锁发送消息，防止对一个session重复发送
     * @param session
     * @param message
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/12 11:07
     */
    private void sendTextWithRetry(Session session, String message) {
        int retries = 0;
        while (retries < 10) {
            try {
                synchronized (session) { // 确保对 session 的写操作是串行化的
                    if (session.isOpen()) {
                        session.getBasicRemote().sendText(message);
                        return; // 成功发送消息后退出
                    }
                }
            } catch (Exception e) {
                System.err.println("发送在线用户广播失败: " + e.getMessage());
            }
            retries++;
            try {
                TimeUnit.MILLISECONDS.sleep(100); // 等待一段时间后重试
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("重试过程中线程被中断: " + e.getMessage());
                break;
            }
        }
        System.err.println("发送在线用户广播失败，达到最大重试次数");
    }

    /**
     * @Description: 异步标记消息为在线
     * @param userId
     * @param role
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/10 11:58
     */
    @Async
    public void markMessagesAsOnline(Long userId, String role) {
        messageMapper.updateMessagesToOnline(userId, role);
    }

    
    /**
     * @Description: 服务端收到客户端消息的方法
     * @param message
     * @param sendSession 发送方的session
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:20
     */
    public void onMessage(String message, Session sendSession) throws IOException {
        //转换成dto
        MessageSendDto dto=JSON.parseObject(message, MessageSendDto.class);
        //验证消息合法性
        if(!validateMessage(dto, sendSession)) {
            sendErrorMessage(sendSession, "消息json数据有误",MessageType.ERROR);
            return;
        }
//        //验证消息源
//        if (!validateMessageSource(dto,sendSession)) {
//            sendErrorMessage(sendSession, "消息json数据有误",MessageType.ERROR);
//            return;
//        }
        //获取接受者的session
        if(Objects.equals(dto.getMessageType(), MessageType.READ)) {
            addUnreadRecordToStream(dto);
            return;
        }
        Session receiveSession = getOnlineUserSession(dto.getToUserId(), dto.getToRole());
        if(receiveSession !=null) {
            //对方在线
            MessageSendResp resp = JSON.parseObject(message, MessageSendResp.class);
            resp.setOnline(true);
            //添加发送方用户相关信息
            addUserRelatedMessageToResp(resp);
            sendTextWithRetry(receiveSession, JSON.toJSONString(resp));
            try {
                saveOlineMessage(dto);
            } catch (Exception e) {
                sendErrorMessage(sendSession, "消息json数据有误",MessageType.ERROR);
            }
        }else {
            //后续离线消息标记
            try {
                saveOfflineMessage(dto);
            } catch (Exception e) {
                sendErrorMessage(sendSession, "消息json数据有误",MessageType.ERROR);
            }
        }
    }

    
    /**
     * @Description: 客户端关闭websocket连接触发的方法
     * @param session 
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:21
     */
    public void onClose(Session session) throws IOException {
        Long userId = getUserIdBySession(session);
        String role = onlineRoles.get(userId);
        if (userId != -1) {
            onlineRoles.remove(userId);
            Session remove = onlineUsers.remove(userId);
            if (remove != null) {
                remove.close();
            }
            session.close();
        }
        log.info("用户关闭连接-"+role+"-"+userId.toString());
        //广播在线用户
        broadcastOnlineUsers();
    }

    
    /**
     * @Description: 删除旧连接，用于异地登录保留新登录的用户
     * @param userId
     * @param newSession 
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:21
     */
    private void deleteOldSession(long userId, Session newSession) {
        for (Map.Entry<Long, Session> entry : onlineUsers.entrySet()) {
            if (entry.getKey().equals(userId) && !entry.getValue().equals(newSession)) {
                Session oldSession = entry.getValue();
                try {
                    sendErrorMessage(oldSession,"异地登录",MessageType.REMOTE_LOGIN_ERROR);
                    oldSession.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "新的连接已建立，旧连接关闭"));
                } catch (IOException e) {
                    e.printStackTrace();
                }
                onlineUsers.remove(userId);
                break;
            }
        }
    }

    /**
     * @Description:  获取用户userId，通过session
     * @param session 
     * @return long
     * @Author lhb
     * @CreateTime 2025/5/27 15:22
     */
    private long getUserIdBySession(Session session) {
        for (Map.Entry<Long, Session> entry : onlineUsers.entrySet()) {
            if (entry.getValue().equals(session)) {
                return entry.getKey();
            }
        }
        return -1; // 如果没有找到对应的userid，返回 -1
    }


    /**
     * @Description: 把握手阶段存储在sec信息返回给前端
     * @param session
     * @param config 
     * @return boolean
     * @Author lhb
     * @CreateTime 2025/5/27 15:22
     */
    private boolean handleHandshakeError(Session session, EndpointConfig config) {
        String handshakeError = (String) config.getUserProperties().get("handshakeError");
        if (handshakeError != null) {
            ErrorMessageResp errorMessageResp=new ErrorMessageResp(handshakeError,MessageType.ERROR,now());
            String errorJson = JSON.toJSONString(errorMessageResp);
            session.getAsyncRemote().sendText(errorJson, result -> {
                if (result.isOK()) {
                    try {
                        session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE,
                                "Handshake error handled"));
                    } catch (IOException e) {
                        // 忽略关闭异常
                    }
                }
            });
            return true;
        }
        return false;
    }



    
    /**
     * @Description: 加工mysql查询的消息，返回排序后的聊天记录
     * @param messages
     * @param userId
     * @return List<LoginMessagesResp>
     * @Author lhb
     * @CreateTime 2025/5/27 15:24
     */
    private List<LoginMessagesResp> processMessages(List<Message> messages, long userId, String role) {
        Set<UserRole> userRoles = extractUserRolesFromMessages(messages, userId, role);
        Map<Long, UserRelatedMessageResp> userRelatedMessageRespMap = generateUserRelatedMessageMap(userRoles, role);
        //根据时间排序
        messages.sort(Comparator.comparing(Message::getCreatedAt));
        //构造聊天记录
        return buildLoginMessages(userRoles, messages, userRelatedMessageRespMap);
    }



    /**
     * @Description: 根据聊天记录获取的消息构造UserRole，这个对象代表着一个用户（通用的userid，可能会有多个角色）
     * @param messages
     * @param userId
     * @return Set<UserRole>
     * @Author lhb
     * @CreateTime 2025/5/27 15:27
     */
    private Set<UserRole> extractUserRolesFromMessages(List<Message> messages, long userId, String role) {
        Set<UserRole> userRoles = new HashSet<>();
        for (Message message : messages) {
            Long contactUserId = message.getFromUserId();
            String contactRole = message.getFromRole();
            if (userId != contactUserId) {
                //userId不同则直接加入关联
                userRoles.add(new UserRole(contactUserId, message.getFromRole()));
            } else {
                //相同需再判断是否是相同角色
                if(contactRole.equals(role)) {
                    userRoles.add(new UserRole(message.getToUserId(), message.getToRole()));
                } else {
                    //说明是同UserId不同角色之间的聊天记录
                    userRoles.add(new UserRole(contactUserId, message.getFromRole()));
                }
            }
        }
        return userRoles;
    }



    /**
     * @Description: 用循环构造聊天记录
     * @param userRoles
     * @param messages
     * @return List<LoginMessagesResp>
     * @Author lhb
     * @CreateTime 2025/5/27 15:29
     */
    private List<LoginMessagesResp> buildLoginMessages(Set<UserRole> userRoles,
                                                       List<Message> messages,
                                                       Map<Long, UserRelatedMessageResp> userRelatedMessageRespMap) {
        List<LoginMessagesResp> loginMessagesRespList = new ArrayList<>();
        for (UserRole userRole : userRoles) {
            LoginMessagesResp loginMessagesResp = new LoginMessagesResp();
            loginMessagesResp.setContactUserId(userRole.getUserId());
            loginMessagesResp.setRole(userRole.getRole());
            //添加用户相关记录
            if(!userRole.getRole().equals(managerRole)) {
                //管理员无需添加
                loginMessagesResp.setUserRelatedMessageResp(userRelatedMessageRespMap.get(userRole.getUserId()));
            }
            //添加与ContactUserId相关的聊天记录
            Iterator<Message> iterator = messages.iterator();
            while (iterator.hasNext()) {
                Message message = iterator.next();
                if ((message.getFromRole().equals(userRole.getRole()) && message.getFromUserId().equals(userRole.getUserId())) ||
                        (message.getToRole().equals(userRole.getRole()) && message.getToUserId().equals(userRole.getUserId()))) {
                    MessageSendResp messageSendResp = new MessageSendResp();
                    //该消息为自己发送的，对该用户来讲不是离线消息
                    if(message.getToUserId().equals(userRole.getUserId())){
                        message.setOnline(true);
                    }
                    BeanUtils.copyProperties(message, messageSendResp);
                    loginMessagesResp.getMessageSendRespList().add(messageSendResp);
                    iterator.remove();
                }
            }
            if (!loginMessagesResp.getMessageSendRespList().isEmpty()) {
                loginMessagesRespList.add(loginMessagesResp);
            }
        }
        return loginMessagesRespList;
    }



    /**
     * @Description: 发送给客户端messageSendResp格式的聊天记录
     * @param session
     * @param userId
     * @param role
     * @param loginMessagesRespList
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:31
     */
    private void sendLoginMessages(Session session, long userId, String role,
                                   List<LoginMessagesResp> loginMessagesRespList) throws IOException {
        MessageSendResp messageSendResp = new MessageSendResp();
        messageSendResp.setMessage(loginMessagesRespList);
        messageSendResp.setToRole(role);
        messageSendResp.setToUserId(userId);
        messageSendResp.setFromUserId(-1L);
        messageSendResp.setFromRole("server");
        messageSendResp.setCreatedAt(now());
        messageSendResp.setMessageType(MessageType.CHAT_RECORD);
        //聊天记录的online字段没有用，但最好不要为空
        messageSendResp.setOnline(true);
        sendTextWithRetry(session,JSON.toJSONString(messageSendResp));
    }



    /**
     * @Description: 验证消息源
     * @param dto
     * @param session 
     * @return boolean
     * @Author lhb
     * @CreateTime 2025/5/27 15:18
     */
    private boolean validateMessageSource(MessageSendDto dto,Session session) {
        Long userId = getUserIdBySession(session);
        if (userId == -1) return false;
        String role = onlineRoles.get(userId);
        return userId.equals(dto.getFromUserId()) && role.equals(dto.getFromRole());
    }
    
    
    /**
     * @Description: 获取在线用户的session
     * @param userId
     * @param role 
     * @return Session
     * @Author lhb
     * @CreateTime 2025/5/27 15:16
     */
    private Session getOnlineUserSession(Long userId,String role){
        String receiveRole = onlineRoles.get(userId);
        Session receiveSession = onlineUsers.get(userId);
        if(receiveSession ==null || receiveRole == null) return null;
        if(!receiveRole.equals(role)) return null;
        return receiveSession;
    }

    
    /**
     * @Description: 保存离线消息到数据库中
     * @param dto 
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:16
     */
    private void saveOfflineMessage(MessageSendDto dto) {
        Message messageDataBase = new Message();
        BeanUtils.copyProperties(dto, messageDataBase);
        messageDataBase.setOnline(false);
        messageMapper.insertUserMessage(messageDataBase);
    }


    /**
     * @Description: 保存在线消息到数据库中
     * @param dto
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/10 11:31
     */
    private void saveOlineMessage(MessageSendDto dto) {
        Message messageDataBase = new Message();
        BeanUtils.copyProperties(dto, messageDataBase);
        messageDataBase.setOnline(true);
        messageMapper.insertUserMessage(messageDataBase);
    }

    
    /**
     * @Description: 往map集合添加用户session
     * @param userId
     * @param session
     * @param role 
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:34
     */
    private void registerUserSession(long userId, Session session, String role) {
        onlineUsers.put(userId, session);
        onlineRoles.put(userId, role);
    }


    /**
     * @Description: 给客户端发送异常信息
     * @param session
     * @param errorMessage 
     * @return void
     * @Author lhb
     * @CreateTime 2025/5/27 15:51
     */
    private void sendErrorMessage(Session session, String errorMessage,MessageType messageType) {
        ErrorMessageResp errorMessageResp=new ErrorMessageResp(errorMessage,messageType,now());
        String errorJson = JSON.toJSONString(errorMessageResp);
        sendTextWithRetry(session,errorJson);
    }

    /**
     * @Description: 验证消息合法性
     * @param dto
     * @param sendSession
     * @return Boolean
     * @Author lhb
     * @CreateTime 2025/7/22 13:50
     */
    private Boolean validateMessage(MessageSendDto dto, Session sendSession) {
        String fromRole = dto.getFromRole();
        String toRole = dto.getToRole();
        //只能管理员直接互相发送，求职者和招聘者之间互相发送
        if((fromRole.equals(managerRole) && toRole.equals(managerRole)) ||
                (fromRole.equals(seekerRole) && toRole.equals(recruiterRole)) ||
                (fromRole.equals(recruiterRole) && toRole.equals(seekerRole))) {
            //校验发送方的session
            return validateMessageSource(dto, sendSession);
        }
        return false;
    }

    /**
     * @Description: 添加用户相关信息至响应体中，用于单发消息
     * @param resp
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/22 14:24
     */
    private void addUserRelatedMessageToResp(MessageSendResp resp) {
        if (resp.getFromRole().equals(managerRole)) {
            //管理员不用设置
            return;
        }
        resp.setUserRelatedMessageResp(new UserRelatedMessageResp());
        User user = userCommonService.getUserByUserId(resp.getFromUserId());
        if(resp.getFromRole().equals(recruiterRole)) {
            Recruiter recruiter = recruiterCrossService.getRecruiterByUserId(resp.getFromUserId());
            Company company = recruiterCrossService.getCompanyMessage(recruiter.getCompanyId());
            resp.getUserRelatedMessageResp().setRole(recruiter.getRole());
            resp.getUserRelatedMessageResp().setCompanyName(company.getName());
        }
        resp.getUserRelatedMessageResp().setName(user.getName());
        resp.getUserRelatedMessageResp().setAvatar(user.getAvatar());
        resp.getUserRelatedMessageResp().setUserId(resp.getFromUserId());
    }

    /**
     * @Description: 批量生成用户相关信息的map，用户聊天记录
     * @param userRoles
     * @param role 
     * @return Map<Long,UserRelatedMessageResp>
     * @Author lhb
     * @CreateTime 2025/7/22 15:52
     */
    private Map<Long, UserRelatedMessageResp> generateUserRelatedMessageMap(Set<UserRole> userRoles, String role) {
        Map<Long, UserRelatedMessageResp> userRelatedMessageRespMap = new HashMap<>();
        if(role.equals(managerRole)) {
            //管理员直接返回空map
            return userRelatedMessageRespMap;
        }
        List<Long> contactUserIds = userRoles.stream()
                .map(UserRole::getUserId)
                .filter(Objects::nonNull)
                .toList();
        List<UserRelatedMessageResp> userRelatedMessageRespList = new ArrayList<>();
        List<User> userList = userCommonService.getUsersByUserIds(contactUserIds);
        if(role.equals(seekerRole)) {
            //批量生成招聘者相关信息
            generateRecruiterRelatedMessage(userList, contactUserIds, userRelatedMessageRespList);
        }
        if(role.equals(recruiterRole)) {
            //批量生成求职者相关信息
            generateSeekerRelatedMessage(userList, userRelatedMessageRespList);
        }
        userRelatedMessageRespMap = userRelatedMessageRespList.stream()
                .collect(Collectors.toMap(UserRelatedMessageResp::getUserId,
                        userRelatedMessageResp -> userRelatedMessageResp));
        return userRelatedMessageRespMap;
    }


    /**
     * @Description: 拆分出来的方法，批量生成求职者相关信息
     * @param userList
     * @param userRelatedMessageRespList
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/22 15:56
     */
    private void generateSeekerRelatedMessage(List<User> userList,
                                                 List<UserRelatedMessageResp> userRelatedMessageRespList) {
        for (User user : userList) {
            UserRelatedMessageResp userRelatedMessageResp = new UserRelatedMessageResp();
            userRelatedMessageResp.setUserId(user.getUserId());
            userRelatedMessageResp.setAvatar(user.getAvatar());
            userRelatedMessageResp.setName(user.getName());
            userRelatedMessageRespList.add(userRelatedMessageResp);
        }
    }


    /**
     * @Description: 拆分出来的方法，批量生成招聘者相关信息
     * @param userList
     * @param contactUserIds
     * @param userRelatedMessageRespList
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/22 15:58
     */
    private void generateRecruiterRelatedMessage(List<User> userList,
                                                 List<Long> contactUserIds,
                                                 List<UserRelatedMessageResp> userRelatedMessageRespList) {
        List<Recruiter> recruiterList = recruiterCrossService.getRecruiterByUserIds(contactUserIds);
        Set<Long> companyIdSet = recruiterList.stream()
                .map(Recruiter::getCompanyId)
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());
        List<Long> companyIds = new ArrayList<>(companyIdSet);
        List<Company> companyList = recruiterCrossService.getCompanyMessagesByIds(companyIds);
        Map<Long, User> userMap = userList.stream()
                .collect(Collectors.toMap(User::getUserId, user -> user));
        Map<Long, Company> companyMap = companyList.stream()
                .collect(Collectors.toMap(Company::getCompanyId, company -> company));
        for (Recruiter recruiter : recruiterList) {
            UserRelatedMessageResp userRelatedMessageResp = new UserRelatedMessageResp();
            User user = userMap.get(recruiter.getUserId());
            Company company = companyMap.get(recruiter.getCompanyId());
            userRelatedMessageResp.setAvatar(user.getAvatar());
            userRelatedMessageResp.setName(user.getName());
            userRelatedMessageResp.setRole(recruiter.getRole());
            userRelatedMessageResp.setCompanyName(company.getName());
            userRelatedMessageResp.setUserId(recruiter.getUserId());
            userRelatedMessageRespList.add(userRelatedMessageResp);
        }
    }

    /**
     * @Description: 清空聊天记录，关闭连接
     * @param userId
     * @param role
     * @return void
     * @Author lhb
     * @CreateTime 2025/7/28 9:32
     */
    public void clearUserMessageRecord(Long userId, String role){
        if(Objects.equals(role, managerRole) || Objects.equals(role, recruiterRole)) {
            String onlineRole = onlineRoles.get(userId);
            if(StringUtils.isNotBlank(onlineRole) && Objects.equals(onlineRole, role)) {
                onlineRoles.remove(userId);
                Session remove = onlineUsers.remove(userId);
                if (remove != null) {
                    try {
                        remove.close();
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
                //广播在线用户
                broadcastOnlineUsers();
            }
            messageMapper.clearUserMessageRecord(userId, role);
        }
    }
}
